
package he.consumer;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import he.service.ProductService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;



@Component
public class ElasticSearchConsumer {

    private final ObjectMapper objectMapper = new ObjectMapper();
    private final ProductService productService;

    @Autowired
    public ElasticSearchConsumer(ProductService productService) {
        this.productService = productService;
    }

    @KafkaListener(topics = "shop-mysql-server.shopdb.product", groupId = "es-group")
    public void consume(String message) {
        try {
            JsonNode jsonNode = objectMapper.readTree(message);
            // 解析变更事件并更新 Elasticsearch
            // 示例：productService.saveProduct(...)
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
